package hbase;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.hbase.bolt.HBaseBolt;
import org.apache.storm.hbase.bolt.mapper.SimpleHBaseMapper;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;
import redis.WordSpout;

import java.util.HashMap;
import java.util.Map;

public class PersistentWordCount {
    public static void main(String[] args) throws Exception {
        Config config = new Config();

        Map<String, Object> hbaseConfig = new HashMap<>();
        hbaseConfig.put("hbase.rootdir", "hdfs://liu:9000/hbase");
        hbaseConfig.put("hbase.zookeeper.quorum", "liu:2181,liu:2182,liu:2183");
        // 将hbase的配置加到storm中
        config.put("hbase.conf", hbaseConfig);

        WordSpout wordSpout = new WordSpout();
        WordCountBolt wordCountBolt = new WordCountBolt();
        SimpleHBaseMapper simpleHBaseMapper = new SimpleHBaseMapper()
                .withColumnFamily("cf")
                .withRowKeyField("word")
                .withColumnFields(new Fields("word"))
                .withCounterFields(new Fields("count"));

        HBaseBolt hBaseBolt = new HBaseBolt("WordCount", simpleHBaseMapper)
                .withConfigKey("hbase.conf"); /*不加可能会连不上*/

        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("wordSpout", wordSpout);
        builder.setBolt("wordCountBolt", wordCountBolt).shuffleGrouping("wordSpout");
        builder.setBolt("hBaseBolt", hBaseBolt).fieldsGrouping("wordCountBolt", new Fields("word"));

        LocalCluster localCluster = new LocalCluster();
        localCluster.submitTopology("hbase-topology", config, builder.createTopology());
    }
}
